package com.zhengbo.flink.demo.base.kafka;

import com.google.common.base.Preconditions;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;

/**
 * @program: mq-flink
 * @description: ${description}
 * @author: maqiang
 * @create: 2019/8/1
 */

public class KafkaConfig {


    /**
     * @param brokerList
     * @param groupId
     * @return
     */
    public static Properties kafkaConsumerProperties(String brokerList, String groupId) {
        Preconditions.checkArgument(!StringUtils.isEmpty(brokerList), "brokerList is empty.");
        Preconditions.checkArgument(!StringUtils.isEmpty(groupId), "topic is empty.");

        Properties kafkaProperties = new Properties();
        kafkaProperties.put("bootstrap.servers", brokerList);
        kafkaProperties.setProperty("group.id", groupId);
        kafkaProperties.put("enable.auto.commit", "true");
        kafkaProperties.put("auto.commit.interval.ms", 1000);
        kafkaProperties.put("session.timeout.ms", 30000);
        kafkaProperties.put("max.poll.records", 100);
        kafkaProperties.put("key.deserializer", StringDeserializer.class.getName());
        kafkaProperties.put("value.deserializer", StringDeserializer.class.getName());
        return kafkaProperties;
    }

    /**
     * @param brokerList
     * @param groupId
     * @return
     */
    public static Properties byteKafkaProps(String brokerList, String groupId) {
        Preconditions.checkArgument(!StringUtils.isEmpty(brokerList), "brokerList is empty.");
        Preconditions.checkArgument(!StringUtils.isEmpty(groupId), "topic is empty.");

        Properties kafkaProperties = new Properties();
        kafkaProperties.put("bootstrap.servers", brokerList);
        kafkaProperties.setProperty("group.id", groupId);
        kafkaProperties.put("enable.auto.commit", "true");
        kafkaProperties.put("auto.commit.interval.ms", 1000);
        kafkaProperties.put("session.timeout.ms", 30000);
        kafkaProperties.put("max.poll.records", 100);
        kafkaProperties.put("key.deserializer", StringDeserializer.class.getName());
        kafkaProperties.put("value.deserializer", ByteArrayDeserializer.class.getName());
        return kafkaProperties;
    }

}
